/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.nifi.processors.huawei.obs;

import com.obs.services.ObsClient;
import com.obs.services.exception.ObsException;
import com.obs.services.model.*;
import org.apache.commons.codec.binary.Base64;
import org.apache.nifi.components.AllowableValue;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.flowfile.attributes.CoreAttributes;
import org.apache.nifi.processor.DataUnit;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processors.huawei.obs.encryption.StandardOBSEncryptionService;
import org.apache.nifi.provenance.ProvenanceEventRecord;
import org.apache.nifi.provenance.ProvenanceEventType;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;

import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.security.SecureRandom;
import java.util.*;

import static org.apache.nifi.processors.huawei.obs.Constants.*;
import static org.apache.nifi.processors.huawei.common.PropertyDescriptors.*;
import static org.junit.jupiter.api.Assertions.*;

/**
 * Provides integration level testing with actual Huawei OBS resources for {@link PutOBSObject} and requires additional configuration and resources to work.
 */
public class ITPutOBSObject extends AbstractOBSIT {

    final static String TEST_PARTSIZE_STRING = "50 mb";
    final static Long   TEST_PARTSIZE_LONG = 50L * 1024L * 1024L;

    final static Long S3_MINIMUM_PART_SIZE = 50L * 1024L * 1024L;
    final static Long S3_MAXIMUM_OBJECT_SIZE = 5L * 1024L * 1024L * 1024L;

    private static String kmsKeyId = "your kmsId in this region";

    private static String randomKeyMaterial = "";

    private static String USER_ID = "**The account ids of the authorized user, separated by commas (,)**";
    private static String OWNER_ID = "**Account ID of Owner**";


    @BeforeAll
    public static void setupClass() {
        byte[] keyRawBytes = new byte[32];
        SecureRandom secureRandom = new SecureRandom();
        secureRandom.nextBytes(keyRawBytes);
        randomKeyMaterial = Base64.encodeBase64String(keyRawBytes);
        System.out.println(randomKeyMaterial);
    }

    @Test
    public void testSimplePut() throws IOException {
        TestRunner runner = initRunner(new PutOBSObject());
        assertTrue(runner.setProperty("x-custom-prop", "hello").isValid());
        for (int i = 0; i < 3; i++) {
            final Map<String, String> attrs = new HashMap<>();
            attrs.put("filename", i + ".txt");
            runner.enqueue(getResourcePath(SAMPLE_FILE_RESOURCE_NAME), attrs);
        }
        runner.run(3);
        runner.assertAllFlowFilesTransferred(PutOBSObject.REL_SUCCESS, 3);
    }

    @Test
    public void testSimplePutEncrypted() throws IOException {
        TestRunner runner = initRunner(new PutOBSObject());
        addEncryptionService(runner, StandardOBSEncryptionService.SSE_KMS, kmsKeyId);

        for (int i = 0; i < 3; i++) {
            final Map<String, String> attrs = new HashMap<>();
            attrs.put("filename", i + ".txt");
            runner.enqueue(getResourcePath(SAMPLE_FILE_RESOURCE_NAME), attrs);
        }
        runner.run(3);
        runner.assertAllFlowFilesTransferred(PutOBSObject.REL_SUCCESS, 3);
        final List<MockFlowFile> ffs = runner.getFlowFilesForRelationship(PutOBSObject.REL_SUCCESS);
        for (MockFlowFile flowFile : ffs) {
            flowFile.assertAttributeEquals(OBS_ENCRYPTION_STRATEGY, StandardOBSEncryptionService.SSE_KMS.getValue());
        }
    }

    @Test
    public void testSimplePutFilenameWithNationalCharacters() throws IOException {
        TestRunner runner = initRunner(new PutOBSObject());

        final Map<String, String> attrs = new HashMap<>();
        attrs.put("filename", "测试文件名.txt");
        runner.enqueue(getResourcePath(SAMPLE_FILE_RESOURCE_NAME), attrs);

        runner.run(1);

        runner.assertAllFlowFilesTransferred(PutOBSObject.REL_SUCCESS, 1);
    }

    // put & fetch & compare
    private void testPutThenFetch(String encryptionStrategy) throws IOException {
        // Put
        TestRunner putRunner = initRunner(new PutOBSObject());

        // Server-side KMS
        if(StandardOBSEncryptionService.SSE_KMS.getValue().equals(encryptionStrategy)){
            addEncryptionService(putRunner, StandardOBSEncryptionService.SSE_KMS, kmsKeyId);
        } else if (StandardOBSEncryptionService.SSE_C.getValue().equals(encryptionStrategy)){
            // Server-side Customer Key
            addEncryptionService(putRunner, StandardOBSEncryptionService.SSE_C, randomKeyMaterial);
        }

        final Map<String, String> attrs = new HashMap<>();
        attrs.put("filename",  "filename-obs.txt");
        putRunner.enqueue(getResourcePath(SAMPLE_FILE_RESOURCE_NAME), attrs);
        putRunner.run();

        putRunner.assertAllFlowFilesTransferred(PutOBSObject.REL_SUCCESS, 1);
        List<MockFlowFile> ffs = putRunner.getFlowFilesForRelationship(PutOBSObject.REL_SUCCESS);
        ffs.get(0).assertAttributeEquals(OBS_ENCRYPTION_STRATEGY, encryptionStrategy);

        // Fetch
        TestRunner fetchRunner = initRunner(new FetchOBSObject());
        // Server-side KMS
        if(StandardOBSEncryptionService.SSE_KMS.getValue().equals(encryptionStrategy)){
            addEncryptionService(fetchRunner, StandardOBSEncryptionService.SSE_KMS, kmsKeyId);
        } else if (StandardOBSEncryptionService.SSE_C.getValue().equals(encryptionStrategy)){
            // Server-side Customer Key
            addEncryptionService(fetchRunner, StandardOBSEncryptionService.SSE_C, randomKeyMaterial);
        }
        fetchRunner.enqueue(new byte[0], attrs);
        fetchRunner.run(1);

        fetchRunner.assertAllFlowFilesTransferred(FetchOBSObject.REL_SUCCESS, 1);
        ffs = fetchRunner.getFlowFilesForRelationship(FetchOBSObject.REL_SUCCESS);
        MockFlowFile ff = ffs.get(0);
        // 比较文件内容是否一致
        ff.assertContentEquals(getFileFromResourceName(SAMPLE_FILE_RESOURCE_NAME));
        ff.assertAttributeEquals(OBS_ENCRYPTION_STRATEGY, StandardOBSEncryptionService.STRATEGY_NAME_SSE_C.equals(encryptionStrategy) ? encryptionStrategy : null);
    }

    @Test
    public void testPutThenFetchWithNONE() throws IOException {
        testPutThenFetch(StandardOBSEncryptionService.STRATEGY_NAME_NONE);
    }

    @Test
    public void testPutThenFetchWithSSE() throws IOException {
        testPutThenFetch(StandardOBSEncryptionService.STRATEGY_NAME_SSE_KMS);
    }

    @Test
    public void testPutThenFetchWithSSEC() throws IOException {
        testPutThenFetch(StandardOBSEncryptionService.STRATEGY_NAME_SSE_C);
    }


    @Test
    public void testPutObsObjectUsingCredentialsProviderService() throws Throwable {
        final TestRunner runner = initCredentialsRunner(new PutOBSObject());
        for (int i = 0; i < 3; i++) {
            final Map<String, String> attrs = new HashMap<>();
            attrs.put("filename", i + ".txt");
            runner.enqueue(getResourcePath(SAMPLE_FILE_RESOURCE_NAME), attrs);
        }
        runner.run(3);

        runner.assertAllFlowFilesTransferred(PutOBSObject.REL_SUCCESS, 3);

    }

    @Test
    public void testMetaData() throws IOException {
        PutOBSObject processor = new PutOBSObject();
        final TestRunner runner = initRunner(new PutOBSObject());
        PropertyDescriptor prop1 = processor.getSupportedDynamicPropertyDescriptor("TEST-PROP-1");
        runner.setProperty(prop1, "TESTING-1-2-3");
        PropertyDescriptor prop2 = processor.getSupportedDynamicPropertyDescriptor("TEST-PROP-2");
        runner.setProperty(prop2, "TESTING-4-5-6");

        final Map<String, String> attrs = new HashMap<>();
        attrs.put("filename", "meta.txt");
        runner.enqueue(getResourcePath(SAMPLE_FILE_RESOURCE_NAME), attrs);

        runner.run();

        runner.assertAllFlowFilesTransferred(PutOBSObject.REL_SUCCESS, 1);
        List<MockFlowFile> flowFiles = runner.getFlowFilesForRelationship(PutOBSObject.REL_SUCCESS);
        MockFlowFile ff1 = flowFiles.get(0);
        for (Map.Entry attrib : ff1.getAttributes().entrySet()) {
            System.out.println(attrib.getKey() + " = " + attrib.getValue());
        }
    }

    @Test
    public void testContentType() throws IOException {
        TestRunner runner = initRunner(new PutOBSObject());
        runner.setProperty(PutOBSObject.CONTENT_TYPE, "text/plain");
        runner.enqueue(getResourcePath(SAMPLE_FILE_RESOURCE_NAME));
        runner.run();
        runner.assertAllFlowFilesTransferred(PutOBSObject.REL_SUCCESS, 1);
        List<MockFlowFile> flowFiles = runner.getFlowFilesForRelationship(PutOBSObject.REL_SUCCESS);
        MockFlowFile ff1 = flowFiles.get(0);
        ff1.assertAttributeEquals(OBS_CONTENT_TYPE, "text/plain");
    }

    @Test
    public void testContentDispositionInline() throws IOException {
        TestRunner runner = initRunner(new PutOBSObject());
        runner.setProperty(PutOBSObject.CONTENT_DISPOSITION, PutOBSObject.CONTENT_DISPOSITION_INLINE);
        runner.enqueue(getResourcePath(SAMPLE_FILE_RESOURCE_NAME));
        runner.run();

        runner.assertAllFlowFilesTransferred(PutOBSObject.REL_SUCCESS, 1);
        List<MockFlowFile> flowFiles = runner.getFlowFilesForRelationship(PutOBSObject.REL_SUCCESS);
        MockFlowFile ff1 = flowFiles.get(0);
        ff1.assertAttributeEquals(OBS_CONTENT_DISPOSITION, PutOBSObject.CONTENT_DISPOSITION_INLINE);
    }

    // ok
    @Test
    public void testContentDispositionNull() throws IOException {
        // Put
        TestRunner runner =initRunner(new PutOBSObject());

        final Map<String, String> attrs = new HashMap<>();
        attrs.put("filename",  "filename-on-s3.txt");
        runner.enqueue(getResourcePath(SAMPLE_FILE_RESOURCE_NAME), attrs);
        runner.run();

        runner.assertAllFlowFilesTransferred(PutOBSObject.REL_SUCCESS, 1);
        List<MockFlowFile> ffs = runner.getFlowFilesForRelationship(PutOBSObject.REL_SUCCESS);

        // Fetch
        TestRunner fetchRunner = initRunner(new FetchOBSObject());
        fetchRunner.enqueue(new byte[0], attrs);
        fetchRunner.run(1);

        fetchRunner.assertAllFlowFilesTransferred(FetchOBSObject.REL_SUCCESS, 1);
        ffs = fetchRunner.getFlowFilesForRelationship(FetchOBSObject.REL_SUCCESS);
        MockFlowFile ff = ffs.get(0);
        ff.assertContentEquals(getFileFromResourceName(SAMPLE_FILE_RESOURCE_NAME));
        ff.assertAttributeNotExists(OBS_CONTENT_DISPOSITION);
    }


    //ok
    @Test
    public void testContentDispositionAttachment() throws IOException {
        TestRunner runner = initRunner(new PutOBSObject());
        runner.setProperty(PutOBSObject.CONTENT_DISPOSITION, PutOBSObject.CONTENT_DISPOSITION_ATTACHMENT);
        runner.enqueue(getResourcePath(SAMPLE_FILE_RESOURCE_NAME));
        runner.run();

        runner.assertAllFlowFilesTransferred(PutOBSObject.REL_SUCCESS, 1);
        List<MockFlowFile> flowFiles = runner.getFlowFilesForRelationship(PutOBSObject.REL_SUCCESS);
        MockFlowFile ff1 = flowFiles.get(0);
        ff1.assertAttributeEquals(OBS_CONTENT_DISPOSITION, "attachment; filename=\"hello.txt\"");
    }

    @Test
    public void testCacheControl() throws IOException {
        TestRunner runner = initRunner(new PutOBSObject());

        runner.setProperty(PutOBSObject.CACHE_CONTROL, "no-cache");
        runner.enqueue(getResourcePath(SAMPLE_FILE_RESOURCE_NAME));

        runner.run();

        runner.assertAllFlowFilesTransferred(PutOBSObject.REL_SUCCESS, 1);
        List<MockFlowFile> flowFiles = runner.getFlowFilesForRelationship(PutOBSObject.REL_SUCCESS);
        MockFlowFile ff1 = flowFiles.get(0);
        ff1.assertAttributeEquals(OBS_CACHE_CONTROL, "no-cache");
    }

    @Test
    public void testPutInFolder() throws IOException {
        TestRunner runner = initRunner(new PutOBSObject());
        assertTrue(runner.setProperty("x-custom-prop", "hello").isValid());
        runner.assertValid();
        final Map<String, String> attrs = new HashMap<>();
        attrs.put("filename", "folder/1.txt");
        runner.enqueue(getResourcePath(SAMPLE_FILE_RESOURCE_NAME), attrs);
        runner.run();
        runner.assertAllFlowFilesTransferred(PutOBSObject.REL_SUCCESS, 1);
    }

    // ok
    @Test
    public void testStorageClasses() throws IOException {
        TestRunner runner = initRunner(new PutOBSObject());

        assertTrue(runner.setProperty("x-custom-prop", "hello").isValid());

        for (StorageClassEnum storageClass : StorageClassEnum.values()) {

            runner.setProperty(PutOBSObject.STORAGE_CLASS, storageClass.name());

            final Map<String, String> attrs = new HashMap<>();
            attrs.put("filename", "testStorageClasses/small_" + storageClass.name() + ".txt");
            runner.enqueue(getResourcePath(SAMPLE_FILE_RESOURCE_NAME), attrs);

            runner.run();

            runner.assertAllFlowFilesTransferred(PutOBSObject.REL_SUCCESS, 1);
            FlowFile file = runner.getFlowFilesForRelationship(PutOBSObject.REL_SUCCESS).get(0);
            assertEquals(storageClass.toString(), file.getAttribute(OBS_STORAGE_CLASS));

            runner.clearTransferState();
        }
    }

    // ok
    @Test
    public void testStorageClassesMultipart() throws IOException {
        TestRunner runner = initRunner(new PutOBSObject());
        runner.setProperty(PutOBSObject.MULTIPART_THRESHOLD, "50 MB");
        runner.setProperty(PutOBSObject.MULTIPART_PART_SIZE, "50 MB");
        assertTrue(runner.setProperty("x-custom-prop", "hello").isValid());

        for (StorageClassEnum storageClass : StorageClassEnum.values()) {
            runner.setProperty(PutOBSObject.STORAGE_CLASS, storageClass.name());
            final Map<String, String> attrs = new HashMap<>();
            attrs.put("filename", "testStorageClasses/large_" + storageClass.name() + ".dat");
            runner.enqueue(new byte[50 * 1024 * 1024 + 1], attrs);
            runner.run();

            runner.assertAllFlowFilesTransferred(PutOBSObject.REL_SUCCESS, 1);
            FlowFile file = runner.getFlowFilesForRelationship(PutOBSObject.REL_SUCCESS).get(0);
            assertEquals(storageClass.toString(), file.getAttribute(OBS_STORAGE_CLASS));
            runner.clearTransferState();
        }
    }

    @Test
    public void testPermissions() throws IOException {
        TestRunner runner = initRunner(new PutOBSObject());

        runner.setProperty(OWNER, OWNER_ID);
        runner.setProperty(READ_USER_LIST, USER_ID);
        runner.setProperty(WRITE_ACL_LIST, USER_ID);
        runner.setProperty(READ_ACL_LIST, USER_ID);
        final Map<String, String> attrs = new HashMap<>();
        attrs.put("filename", "4.txt");
        runner.enqueue(getResourcePath(SAMPLE_FILE_RESOURCE_NAME), attrs);

        runner.run();

        runner.assertAllFlowFilesTransferred(PutOBSObject.REL_SUCCESS, 1);
    }

    @Test
    public void testDynamicProperty() throws IOException {
        final String DYNAMIC_ATTRIB_KEY = "fs.runTimestamp";
        final String DYNAMIC_ATTRIB_VALUE = "${now():toNumber()}";


        final TestRunner runner = initRunner(new PutOBSObject());
        final PutOBSObject processor = (PutOBSObject)runner.getProcessor();
        runner.setProperty(PutOBSObject.MULTIPART_PART_SIZE, TEST_PARTSIZE_STRING);
        PropertyDescriptor testAttrib = processor.getSupportedDynamicPropertyDescriptor(DYNAMIC_ATTRIB_KEY);
        runner.setProperty(testAttrib, DYNAMIC_ATTRIB_VALUE);

        final String FILE1_NAME = "file1";
        Map<String, String> attribs = new HashMap<>();
        attribs.put(CoreAttributes.FILENAME.key(), FILE1_NAME);
        runner.enqueue("123".getBytes(), attribs);

        runner.assertValid();
        processor.getPropertyDescriptor(DYNAMIC_ATTRIB_KEY);
        runner.run();
        runner.assertAllFlowFilesTransferred(PutOBSObject.REL_SUCCESS);
        final List<MockFlowFile> successFiles = runner.getFlowFilesForRelationship(PutOBSObject.REL_SUCCESS);
        assertEquals(1, successFiles.size());
        MockFlowFile ff1 = successFiles.get(0);

        Long now = System.currentTimeMillis();
        String millisNow = Long.toString(now);
        String millisOneSecAgo = Long.toString(now - 1000L);
        String usermeta = ff1.getAttribute(OBS_USER_META);
        String[] usermetaLine0 = usermeta.split(System.lineSeparator())[0].split("=");
        String usermetaKey0 = usermetaLine0[0];
        String usermetaValue0 = usermetaLine0[1];
        assertEquals(DYNAMIC_ATTRIB_KEY, usermetaKey0);
        assertTrue(usermetaValue0.compareTo(millisOneSecAgo) >=0 && usermetaValue0.compareTo(millisNow) <= 0);
    }

    @Test
    public void testProvenance() throws InitializationException {
        final String PROV1_FILE = "provfile1";

        final TestRunner runner = initRunner(new PutOBSObject());

        runner.setProperty(KEY, "${filename}");

        Map<String, String> attribs = new HashMap<>();
        attribs.put(CoreAttributes.FILENAME.key(), PROV1_FILE);
        runner.enqueue("prov1 contents".getBytes(), attribs);

        runner.assertValid();
        runner.run();
        runner.assertAllFlowFilesTransferred(PutOBSObject.REL_SUCCESS);
        final List<MockFlowFile> successFiles = runner.getFlowFilesForRelationship(PutOBSObject.REL_SUCCESS);
        assertEquals(1, successFiles.size());

        final List<ProvenanceEventRecord> provenanceEvents = runner.getProvenanceEvents();
        assertEquals(1, provenanceEvents.size());
        ProvenanceEventRecord provRec1 = provenanceEvents.get(0);
        assertEquals(ProvenanceEventType.SEND, provRec1.getEventType());
        assertEquals(runner.getProcessor().getIdentifier(), provRec1.getComponentId());
        assertEquals(10, provRec1.getUpdatedAttributes().size());
        assertEquals(bucketName, provRec1.getUpdatedAttributes().get(OBS_BUCKET));
    }

    @Test
    public void testStateDefaults() {
        PutOBSObject.MultipartState state1 = new PutOBSObject.MultipartState();
        assertEquals(state1.getUploadId(), "");
        assertEquals(state1.getFilePosition(), (Long) 0L);
        assertEquals(state1.getPartETags().size(), 0L);
        assertEquals(state1.getPartSize(), (Long) 0L);
        assertEquals(state1.getStorageClass().toString(), StorageClassEnum.STANDARD.toString());
        assertEquals(state1.getContentLength(), (Long) 0L);
    }

    @Test
    public void testMultipartProperties() throws IOException {
        final TestRunner runner = initRunner(new PutOBSObject());
        final ProcessContext context = runner.getProcessContext();
        runner.setProperty(FULL_CONTROL_USER_LIST, USER_ID);
        runner.setProperty(PutOBSObject.MULTIPART_PART_SIZE, TEST_PARTSIZE_STRING);

        runner.setProperty(KEY, AbstractOBSIT.SAMPLE_FILE_RESOURCE_NAME);

        assertEquals(bucketName, context.getProperty(BUCKET).toString());
        assertEquals(SAMPLE_FILE_RESOURCE_NAME, context.getProperty(KEY).getValue());
        assertEquals(TEST_PARTSIZE_LONG.longValue(), context.getProperty(PutOBSObject.MULTIPART_PART_SIZE).asDataSize(DataUnit.B).longValue());
    }

    @Test
    public void testLocalStatePersistence() throws IOException {
        final PutOBSObject processor = new PutOBSObject();
        final TestRunner runner = TestRunners.newTestRunner(processor);
        runner.setValidateExpressionUsage(false);
        final String bucket = runner.getProcessContext().getProperty(BUCKET).getValue();
        final String key = runner.getProcessContext().getProperty(KEY).getValue();
        final String cacheKey1 = runner.getProcessor().getIdentifier() + "/" + bucket + "/" + key;
        final String cacheKey2 = runner.getProcessor().getIdentifier() + "/" + bucket + "/" + key + "-v2";
        final String cacheKey3 = runner.getProcessor().getIdentifier() + "/" + bucket + "/" + key + "-v3";

        /*
         * store 3 versions of state
         */
        PutOBSObject.MultipartState state1orig = new PutOBSObject.MultipartState();
        processor.persistLocalState(cacheKey1, state1orig);

        PutOBSObject.MultipartState state2orig = new PutOBSObject.MultipartState();
        state2orig.setUploadId("1234");
        state2orig.setContentLength(1234L);
        processor.persistLocalState(cacheKey2, state2orig);

        PutOBSObject.MultipartState state3orig = new PutOBSObject.MultipartState();
        state3orig.setUploadId("5678");
        state3orig.setContentLength(5678L);
        processor.persistLocalState(cacheKey3, state3orig);

        final List<MultipartUpload> uploadList = new ArrayList<>();
        final MultipartUpload upload1 = new MultipartUpload("", key,new Date(), StorageClassEnum.STANDARD, null, null);
        uploadList.add(upload1);
        final MultipartUpload upload2 = new MultipartUpload("1234", key + "-v2",new Date(), StorageClassEnum.STANDARD, null, null);
        uploadList.add(upload2);
        final MultipartUpload upload3 = new MultipartUpload("5678", key + "-v3",new Date(), StorageClassEnum.STANDARD, null, null);
        uploadList.add(upload3);
        final MultipartUploadListing uploadListing = new MultipartUploadListing.Builder().multipartTaskList(uploadList).builder();
        final MockAmazonS3Client mockClient = new MockAmazonS3Client(OBSRegions.fromName(REGION_NAME).getEndpoint());
        mockClient.setListing(uploadListing);

        /*
         * reload and validate stored state
         */
        final PutOBSObject.MultipartState state1new = processor.getLocalStateIfInOBS(mockClient, bucket, cacheKey1);
        assertEquals("", state1new.getUploadId());
        assertEquals(0L, state1new.getFilePosition().longValue());
        assertEquals(new ArrayList<PartEtag>(), state1new.getPartETags());
        assertEquals(0L, state1new.getPartSize().longValue());
        assertEquals(StorageClassEnum.STANDARD.toString(), state1new.getStorageClass().getCode());
        assertEquals(0L, state1new.getContentLength().longValue());

        final PutOBSObject.MultipartState state2new = processor.getLocalStateIfInOBS(mockClient, bucket, cacheKey2);
        assertEquals("1234", state2new.getUploadId());
        assertEquals(0L, state2new.getFilePosition().longValue());
        assertEquals(new ArrayList<PartEtag>(), state2new.getPartETags());
        assertEquals(0L, state2new.getPartSize().longValue());
        assertEquals(StorageClassEnum.STANDARD.toString(), state2new.getStorageClass().getCode());
        assertEquals(1234L, state2new.getContentLength().longValue());

        final PutOBSObject.MultipartState state3new = processor.getLocalStateIfInOBS(mockClient, bucket, cacheKey3);
        assertEquals("5678", state3new.getUploadId());
        assertEquals(0L, state3new.getFilePosition().longValue());
        assertEquals(new ArrayList<PartEtag>(), state3new.getPartETags());
        assertEquals(0L, state3new.getPartSize().longValue());
        assertEquals(StorageClassEnum.STANDARD.toString(), state3new.getStorageClass().getCode());
        assertEquals(5678L, state3new.getContentLength().longValue());
    }

    @Test
    public void testStatePersistsETags() throws IOException {
        final PutOBSObject processor = new PutOBSObject();
        final TestRunner runner = TestRunners.newTestRunner(processor);
        runner.setValidateExpressionUsage(false);
        final String bucket = runner.getProcessContext().getProperty(BUCKET).getValue();
        final String key = runner.getProcessContext().getProperty(KEY).getValue();
        final String cacheKey1 = runner.getProcessor().getIdentifier() + "/" + bucket + "/" + key + "-bv1";
        final String cacheKey2 = runner.getProcessor().getIdentifier() + "/" + bucket + "/" + key + "-bv2";
        final String cacheKey3 = runner.getProcessor().getIdentifier() + "/" + bucket + "/" + key + "-bv3";

        /*
         * store 3 versions of state
         */
        PutOBSObject.MultipartState state1orig = new PutOBSObject.MultipartState();
        processor.persistLocalState(cacheKey1, state1orig);

        PutOBSObject.MultipartState state2orig = new PutOBSObject.MultipartState();
        state2orig.setUploadId("1234");
        state2orig.setContentLength(1234L);
        processor.persistLocalState(cacheKey2, state2orig);

        PutOBSObject.MultipartState state3orig = new PutOBSObject.MultipartState();
        state3orig.setUploadId("5678");
        state3orig.setContentLength(5678L);
        processor.persistLocalState(cacheKey3, state3orig);

        /*
         * persist state to caches so that
         *      1. v2 has 2 and then 4 tags
         *      2. v3 has 4 and then 2 tags
         */
        state2orig.getPartETags().add(new PartEtag("state 2 tag one", 1));
        state2orig.getPartETags().add(new PartEtag("state 2 tag two", 2));
        processor.persistLocalState(cacheKey2, state2orig);
        state2orig.getPartETags().add(new PartEtag("state 2 tag three", 3));
        state2orig.getPartETags().add(new PartEtag("state 2 tag four", 4));
        processor.persistLocalState(cacheKey2, state2orig);

        state3orig.getPartETags().add(new PartEtag("state 3 tag one", 1));
        state3orig.getPartETags().add(new PartEtag("state 3 tag two", 2));
        state3orig.getPartETags().add(new PartEtag("state 3 tag three", 3));
        state3orig.getPartETags().add(new PartEtag("state 3 tag four", 4));
        processor.persistLocalState(cacheKey3, state3orig);
        state3orig.getPartETags().remove(state3orig.getPartETags().size() - 1);
        state3orig.getPartETags().remove(state3orig.getPartETags().size() - 1);
        processor.persistLocalState(cacheKey3, state3orig);

        final List<MultipartUpload> uploadList = new ArrayList<>();
        final MultipartUpload upload1 = new MultipartUpload("1234", key + "-bv2",new Date(), StorageClassEnum.STANDARD, null, null);
        uploadList.add(upload1);
        final MultipartUpload upload2 = new MultipartUpload("5678", key + "-bv3",new Date(), StorageClassEnum.STANDARD, null, null);
        uploadList.add(upload2);
        final MultipartUploadListing uploadListing = new MultipartUploadListing.Builder().multipartTaskList(uploadList).builder();
        final MockAmazonS3Client mockClient = new MockAmazonS3Client(OBSRegions.fromName(REGION_NAME).getEndpoint());
        mockClient.setListing(uploadListing);

        /*
         * load state and validate that
         *     1. v2 restore shows 4 tags
         *     2. v3 restore shows 2 tags
         */
        final PutOBSObject.MultipartState state2new = processor.getLocalStateIfInOBS(mockClient, bucket, cacheKey2);
        assertEquals("1234", state2new.getUploadId());
        assertEquals(4, state2new.getPartETags().size());

        final PutOBSObject.MultipartState state3new = processor.getLocalStateIfInOBS(mockClient, bucket, cacheKey3);
        assertEquals("5678", state3new.getUploadId());
        assertEquals(2, state3new.getPartETags().size());
    }

    @Test
    public void testStateRemove() throws IOException {
        final PutOBSObject processor = new PutOBSObject();
        final TestRunner runner = TestRunners.newTestRunner(processor);
        runner.setValidateExpressionUsage(false);
        final String bucket = runner.getProcessContext().getProperty(BUCKET).getValue();
        final String key = runner.getProcessContext().getProperty(KEY).getValue();
        final String cacheKey = runner.getProcessor().getIdentifier() + "/" + bucket + "/" + key + "-sr";

        final List<MultipartUpload> uploadList = new ArrayList<>();
        final MultipartUpload upload1 = new MultipartUpload("1234", key ,new Date(), StorageClassEnum.STANDARD, null, null);
        uploadList.add(upload1);
        final MultipartUploadListing uploadListing = new MultipartUploadListing.Builder().multipartTaskList(uploadList).builder();
        final MockAmazonS3Client mockClient = new MockAmazonS3Client(OBSRegions.fromName(REGION_NAME).getEndpoint());
        mockClient.setListing(uploadListing);

        /*
         * store state, retrieve and validate, remove and validate
         */
        PutOBSObject.MultipartState stateOrig = new PutOBSObject.MultipartState();
        stateOrig.setUploadId("1234");
        stateOrig.setContentLength(1234L);
        processor.persistLocalState(cacheKey, stateOrig);

        PutOBSObject.MultipartState state1 = processor.getLocalStateIfInOBS(mockClient, bucket, cacheKey);
        assertEquals("1234", state1.getUploadId());
        assertEquals(1234L, state1.getContentLength().longValue());

        processor.persistLocalState(cacheKey, null);
        PutOBSObject.MultipartState state2 = processor.getLocalStateIfInOBS(mockClient, bucket, cacheKey);
        assertNull(state2);
    }

    @Test
    public void testMultipartSmallerThanMinimum() throws IOException {
        final String FILE1_NAME = "file1";

        final byte[] megabyte = new byte[1024 * 1024];
        final Path tempFile = Files.createTempFile("s3mulitpart", "tmp");
        final FileOutputStream tempOut = new FileOutputStream(tempFile.toFile());
        long tempByteCount = 0;
        for (int i = 0; i < 5; i++) {
            tempOut.write(megabyte);
            tempByteCount += megabyte.length;
        }
        tempOut.close();
        System.out.println("file size: " + tempByteCount);
        assertTrue(tempByteCount < S3_MINIMUM_PART_SIZE);

        assertTrue(megabyte.length < S3_MINIMUM_PART_SIZE);
        assertTrue(TEST_PARTSIZE_LONG >= S3_MINIMUM_PART_SIZE && TEST_PARTSIZE_LONG <= S3_MAXIMUM_OBJECT_SIZE);

        final TestRunner runner = initRunner(new PutOBSObject());
        runner.setProperty(PutOBSObject.MULTIPART_PART_SIZE, TEST_PARTSIZE_STRING);

        Map<String, String> attributes = new HashMap<>();
        attributes.put(CoreAttributes.FILENAME.key(), FILE1_NAME);
        runner.enqueue(new FileInputStream(tempFile.toFile()), attributes);

        runner.assertValid();
        runner.run();
        runner.assertAllFlowFilesTransferred(PutOBSObject.REL_SUCCESS);
        final List<MockFlowFile> successFiles = runner.getFlowFilesForRelationship(PutOBSObject.REL_SUCCESS);
        assertEquals(1, successFiles.size());
        final List<MockFlowFile> failureFiles = runner.getFlowFilesForRelationship(PutOBSObject.REL_FAILURE);
        assertEquals(0, failureFiles.size());
        MockFlowFile ff1 = successFiles.get(0);
        assertEquals(OBS_API_METHOD_PUT_OBJECT, ff1.getAttribute(OBS_API_METHOD_ATTR_KEY));
        assertEquals(FILE1_NAME, ff1.getAttribute(CoreAttributes.FILENAME.key()));
        assertEquals(bucketName, ff1.getAttribute(OBS_BUCKET));
        assertEquals(FILE1_NAME, ff1.getAttribute(OBS_OBJECT));
        assertEquals(tempByteCount, ff1.getSize());
    }

    @Test
    public void testMultipartBetweenMinimumAndMaximum() throws IOException {
        final String FILE1_NAME = "file1";

        final byte[] megabyte = new byte[1024 * 1024];
        final Path tempFile = Files.createTempFile("s3mulitpart", "tmp");
        final FileOutputStream tempOut = new FileOutputStream(tempFile.toFile());
        long tempByteCount = 0;
        for ( ; tempByteCount < TEST_PARTSIZE_LONG + 1; ) {
            tempOut.write(megabyte);
            tempByteCount += megabyte.length;
        }
        tempOut.close();
        System.out.println("file size: " + tempByteCount);
        assertTrue(tempByteCount > S3_MINIMUM_PART_SIZE && tempByteCount < S3_MAXIMUM_OBJECT_SIZE);
        assertTrue(tempByteCount > TEST_PARTSIZE_LONG);

        final TestRunner runner = initRunner(new PutOBSObject());
        runner.setProperty(PutOBSObject.MULTIPART_THRESHOLD, TEST_PARTSIZE_STRING);
        runner.setProperty(PutOBSObject.MULTIPART_PART_SIZE, TEST_PARTSIZE_STRING);

        Map<String, String> attributes = new HashMap<>();
        attributes.put(CoreAttributes.FILENAME.key(), FILE1_NAME);
        runner.enqueue(new FileInputStream(tempFile.toFile()), attributes);

        runner.assertValid();
        runner.run();
        runner.assertAllFlowFilesTransferred(PutOBSObject.REL_SUCCESS);
        final List<MockFlowFile> successFiles = runner.getFlowFilesForRelationship(PutOBSObject.REL_SUCCESS);
        assertEquals(1, successFiles.size());
        final List<MockFlowFile> failureFiles = runner.getFlowFilesForRelationship(PutOBSObject.REL_FAILURE);
        assertEquals(0, failureFiles.size());
        MockFlowFile ff1 = successFiles.get(0);
        assertEquals(OBS_API_METHOD_MULTIPART_UPLOAD, ff1.getAttribute(OBS_API_METHOD_ATTR_KEY));
        assertEquals(FILE1_NAME, ff1.getAttribute(CoreAttributes.FILENAME.key()));
        assertEquals(bucketName, ff1.getAttribute(OBS_BUCKET));
        assertEquals(FILE1_NAME, ff1.getAttribute(OBS_OBJECT));
        assertEquals(tempByteCount, ff1.getSize());
    }

    @Test
    public void testMultipartLargerThanObjectMaximum() throws IOException {
        final String FILE1_NAME = "file1";

        final byte[] megabyte = new byte[1024 * 1024];
        final Path tempFile = Files.createTempFile("s3mulitpart", "tmp");
        final FileOutputStream tempOut = new FileOutputStream(tempFile.toFile());
        for (int i = 0; i < 305; i++) {
            tempOut.write(megabyte);
        }
        tempOut.close();

        final TestRunner runner = initRunner(new PutOBSObject());
        runner.setProperty(PutOBSObject.MULTIPART_PART_SIZE, TEST_PARTSIZE_STRING);
        runner.setProperty(PutOBSObject.MULTIPART_THRESHOLD, "280 mb");

        Map<String, String> attributes = new HashMap<>();
        attributes.put(CoreAttributes.FILENAME.key(), FILE1_NAME);
        runner.enqueue(new FileInputStream(tempFile.toFile()), attributes);

        runner.assertValid();
        runner.run();
        runner.assertAllFlowFilesTransferred(PutOBSObject.REL_SUCCESS);
        final List<MockFlowFile> successFiles = runner.getFlowFilesForRelationship(PutOBSObject.REL_SUCCESS);
        assertEquals(1, successFiles.size());
        final List<MockFlowFile> failureFiles = runner.getFlowFilesForRelationship(PutOBSObject.REL_FAILURE);
        assertEquals(0, failureFiles.size());
        MockFlowFile ff1 = successFiles.get(0);
        assertEquals(FILE1_NAME, ff1.getAttribute(CoreAttributes.FILENAME.key()));
        assertEquals(bucketName, ff1.getAttribute(OBS_BUCKET));
        assertEquals(FILE1_NAME, ff1.getAttribute(OBS_OBJECT));
        assertTrue(ff1.getSize() == 305 * 1024 * 1024);
    }

    @Test
    public void testS3MultipartAgeoff() throws InterruptedException, IOException {
        final TestRunner runner = initRunner(new PutOBSObject());
        final ProcessContext context = runner.getProcessContext();
        final PutOBSObject processor = (PutOBSObject)runner.getProcessor();
        // set check interval and age off to minimum values
        runner.setProperty(PutOBSObject.MULTIPART_AGEOFF_INTERVAL, "1 milli");
        runner.setProperty(PutOBSObject.MULTIPART_MAX_AGE, "1 milli");

        // create some dummy uploads
        for (Integer i = 0; i < 3; i++) {
            final InitiateMultipartUploadRequest initiateRequest = new InitiateMultipartUploadRequest(
                    bucketName, "file" + i + ".txt");
            assertDoesNotThrow(() -> client.initiateMultipartUpload(initiateRequest));
        }

        // Age off is time dependent, so test has some timing constraints.  This
        // sleep() delays long enough to satisfy interval and age intervals.
        Thread.sleep(2000L);

        // actually initialize the state
        MultipartUploadListing uploadList = processor.getAgeOffListAndAgeOffLocalState(context, client, System.currentTimeMillis(), bucketName);
        assertEquals(3, uploadList.getMultipartTaskList().size());

        Thread.sleep(500L);

        // abort a task
        MultipartUpload upload0 = uploadList.getMultipartTaskList().get(0);
        processor.abortMultipartUpload(client, bucketName, upload0);
        uploadList = processor.getAgeOffListAndAgeOffLocalState(context, client, System.currentTimeMillis(), bucketName);
        // ageOff list after abort a task
        assertEquals(2, uploadList.getMultipartTaskList().size());

        Thread.sleep(500L);

        final Map<String, String> attrs = new HashMap<>();
        attrs.put("filename", "test-upload.txt");
        runner.enqueue(getResourcePath(SAMPLE_FILE_RESOURCE_NAME), attrs);
        runner.run();

        uploadList = processor.getAgeOffListAndAgeOffLocalState(context, client, System.currentTimeMillis(), bucketName);
        assertEquals(0, uploadList.getMultipartTaskList().size());
    }

    @Test
    public void testEncryptionServiceWithServerSideKMSEncryptionStrategyUsingSingleUpload() throws IOException, InitializationException {
        byte[] smallData = Files.readAllBytes(getResourcePath(SAMPLE_FILE_RESOURCE_NAME));
        testEncryptionServiceWithServerSideKMSEncryptionStrategy(smallData);
    }

    @Test
    public void testEncryptionServiceWithServerSideKMSEncryptionStrategyUsingMultipartUpload() throws IOException, InitializationException {
        byte[] largeData = new byte[51 * 1024 * 1024];
        testEncryptionServiceWithServerSideKMSEncryptionStrategy(largeData);
    }

    private void testEncryptionServiceWithServerSideKMSEncryptionStrategy(byte[] data) throws IOException, InitializationException {
        TestRunner runner = createPutEncryptionTestRunner(StandardOBSEncryptionService.SSE_KMS, kmsKeyId);

        final Map<String, String> attrs = new HashMap<>();
        attrs.put("filename", "test.txt");
        runner.enqueue(data, attrs);
        runner.assertValid();
        runner.run();
        runner.assertAllFlowFilesTransferred(PutOBSObject.REL_SUCCESS);

        List<MockFlowFile> flowFiles = runner.getFlowFilesForRelationship(PutOBSObject.REL_SUCCESS);
        assertEquals(1, flowFiles.size());
        assertEquals(0, runner.getFlowFilesForRelationship(PutOBSObject.REL_FAILURE).size());
        MockFlowFile putSuccess = flowFiles.get(0);
        assertEquals(putSuccess.getAttribute(OBS_ENCRYPTION_STRATEGY), ObsServiceEncryptionService.STRATEGY_NAME_SSE_KMS);

        MockFlowFile flowFile = fetchEncryptedFlowFile(attrs, StandardOBSEncryptionService.SSE_KMS, kmsKeyId);
        flowFile.assertContentEquals(data);
        flowFile.assertAttributeEquals(OBS_SSE_ALGORITHM, null);
        flowFile.assertAttributeEquals(OBS_ENCRYPTION_STRATEGY, null);
    }

    @Test
    public void testEncryptionServiceWithServerSideCEncryptionStrategyUsingSingleUpload() throws IOException, InitializationException {
        byte[] smallData = Files.readAllBytes(getResourcePath(SAMPLE_FILE_RESOURCE_NAME));
        testEncryptionServiceWithServerSideCEncryptionStrategy(smallData);
    }

    @Test
    public void testEncryptionServiceWithServerSideCEncryptionStrategyUsingMultipartUpload() throws IOException, InitializationException {
        byte[] largeData = new byte[51 * 1024 * 1024];
        testEncryptionServiceWithServerSideCEncryptionStrategy(largeData);
    }

    private void testEncryptionServiceWithServerSideCEncryptionStrategy(byte[] data) throws IOException, InitializationException {
        TestRunner runner = createPutEncryptionTestRunner(StandardOBSEncryptionService.SSE_C, randomKeyMaterial);

        final Map<String, String> attrs = new HashMap<>();
        attrs.put("filename", "test.txt");
        runner.enqueue(data, attrs);
        runner.assertValid();
        runner.run();
        runner.assertAllFlowFilesTransferred(PutOBSObject.REL_SUCCESS);

        List<MockFlowFile> flowFiles = runner.getFlowFilesForRelationship(PutOBSObject.REL_SUCCESS);
        assertEquals(1, flowFiles.size());
        assertEquals(0, runner.getFlowFilesForRelationship(PutOBSObject.REL_FAILURE).size());
        MockFlowFile putSuccess = flowFiles.get(0);
        assertEquals(putSuccess.getAttribute(OBS_ENCRYPTION_STRATEGY), ObsServiceEncryptionService.STRATEGY_NAME_SSE_C);

        MockFlowFile flowFile = fetchEncryptedFlowFile(attrs, StandardOBSEncryptionService.SSE_C, randomKeyMaterial);
        flowFile.assertContentEquals(data);
        // successful fetch does not indicate type of original encryption:
        flowFile.assertAttributeEquals(OBS_SSE_ALGORITHM, SSEAlgorithmEnum.AES256.getCode());
        // but it does indicate it via our specific attribute:
        flowFile.assertAttributeEquals(OBS_ENCRYPTION_STRATEGY, ObsServiceEncryptionService.STRATEGY_NAME_SSE_C);
    }

    private TestRunner createPutEncryptionTestRunner(AllowableValue encryptionStrategy, String keyIdOrMaterial) throws InitializationException {
        TestRunner runner = initRunner(new PutOBSObject());
        addEncryptionService(runner, encryptionStrategy, keyIdOrMaterial);

        runner.setProperty(PutOBSObject.MULTIPART_THRESHOLD, "50 MB");
        runner.setProperty(PutOBSObject.MULTIPART_PART_SIZE, "50 MB");

        return runner;
    }

    private MockFlowFile fetchEncryptedFlowFile(Map<String, String> attributes, AllowableValue encryptionStrategy, String keyIdOrMaterial) throws InitializationException {
        TestRunner runner = initRunner(new FetchOBSObject());
        if (StandardOBSEncryptionService.SSE_C.equals(encryptionStrategy)) {
            addEncryptionService(runner, encryptionStrategy, keyIdOrMaterial);
        }
        runner.enqueue(new byte[0], attributes);
        runner.run(1);
        runner.assertAllFlowFilesTransferred(FetchOBSObject.REL_SUCCESS, 1);
        List<MockFlowFile> flowFiles = runner.getFlowFilesForRelationship(FetchOBSObject.REL_SUCCESS);
        return flowFiles.get(0);
    }


    private class MockAmazonS3Client extends ObsClient {
        MultipartUploadListing listing;

        public MockAmazonS3Client(String endPoint) {
            super(endPoint);
        }

        public void setListing(MultipartUploadListing newlisting) {
            listing = newlisting;
        }

        @Override
        public MultipartUploadListing listMultipartUploads(ListMultipartUploadsRequest listMultipartUploadsRequest)
                throws ObsException {
            return listing;
        }
    }
}
